package io.kiki.sba.registry.server.shared.meta;


import io.kiki.sba.registry.api.exchange.message.Response;
import io.kiki.sba.registry.common.model.ClientInterfaceId;
import io.kiki.sba.registry.common.model.GenericResponse;
import io.kiki.sba.registry.common.model.Node.NodeType;
import io.kiki.sba.registry.common.model.elector.Leader;
import io.kiki.sba.registry.common.model.metaserver.*;
import io.kiki.sba.registry.common.model.metaserver.blacklist.RegistryForbiddenServerRequest;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartBeatResponse;
import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest;
import io.kiki.sba.registry.common.model.metaserver.nodes.ClientInterfaceNode;
import io.kiki.sba.registry.common.model.slot.GetSlotsStatusRequest;
import io.kiki.sba.registry.common.model.slot.SlotsStatusResponse;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.shared.config.CommonConfig;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.StringFormatter;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.atomic.AtomicInteger;


public abstract class AbstractMetaServerService<T extends HeartBeatResponse> implements MetaServerService {
    static final int MAX_RENEW_FAIL_COUNT = 3;
    protected final Logger logger = LoggerFactory.getLogger(AbstractMetaServerService.class);
    final Renewer renewer = new Renewer();
    final AtomicInteger renewFailCounter = new AtomicInteger(0);
    @Autowired
    protected CommonConfig commonConfig;
    protected volatile State state = State.NULL;
    @Autowired
    private CentralControlLeaderExchanger centralControlLeaderExchanger;
    private Thread renewerThread;

    @Override
    public synchronized void startRenewer() {
        if (renewerThread == null) {
            renewerThread = ConcurrentUtils.createDaemonThread("meta-renewer", this.renewer);
            renewerThread.start();
        }
    }

    @Override
    public void suspendRenewer() {
        renewer.suspend();
        logger.info("suspend the renewer");
    }

    @Override
    public void resumeRenewer() {
        renewer.resume();
        logger.info("resume the renewer");
    }

    public abstract int getRenewIntervalSecs();

    @Override
    public boolean handleSlotsChange(SlotsChangeEvent event) {
        long epoch = event.getSlotsEpoch();
        long currentEpoch = getCurrentslotsEpoch();
        if (currentEpoch >= epoch) {
            logger.warn("[handleSlotsChange] slot-table change event epoch: [{}], current epoch: [{}], " + "won't retrieve again", epoch, currentEpoch);
            return false;
        }
        logger.info("[handleSlotsChange] slot table is changed, run heart-beat to retrieve new version");
        renewer.wakeup();
        return true;
    }

    @Override
    public void addSelfToMetaBlacklist() {
        centralControlLeaderExchanger.sendRequest(commonConfig.getDataCenter(), new RegistryForbiddenServerRequest(DataOperation.add, nodeType(), ServerEnv.IP, cell()));
    }

    @Override
    public void removeSelfFromMetaBlacklist() {
        centralControlLeaderExchanger.sendRequest(commonConfig.getDataCenter(), new RegistryForbiddenServerRequest(DataOperation.remove, nodeType(), ServerEnv.IP, cell()));
    }

    @Override
    public boolean renewNode() {
        final String leaderIp = getMetaServerLeader();
        final long startTimestamp = System.currentTimeMillis();
        boolean success = true;
        try {
            HeartbeatRequest heartbeatRequest = createRequest();
            GenericResponse<T> genericResponse = (GenericResponse<T>) centralControlLeaderExchanger.sendRequest(commonConfig.getDataCenter(), heartbeatRequest).getResult();
            handleHeartbeatResponse(genericResponse);

            success = true;
        } catch (Throwable e) {
            success = false;
            handleHeartbeatFailed(leaderIp, e);
        } finally {
            logger.info("[renewMetaLeader]{},leader={},span={}", success ? 'Y' : 'N', leaderIp, System.currentTimeMillis() - startTimestamp);
        }
        return success;
    }

    boolean checkRenewFailCounter() {
        if (renewFailCounter.get() >= MAX_RENEW_FAIL_COUNT) {
            logger.error("renewNode failed [{}] times, prepare to reset leader from rest api.", renewFailCounter.get());
            centralControlLeaderExchanger.resetLeader(commonConfig.getDataCenter());
            renewFailCounter.set(0);
            return true;
        }
        return false;
    }

    void handleHeartbeatResponse(GenericResponse<T> genericResponse) {
        if (genericResponse == null) {
            throw new RuntimeException("renew node to metaServer error : resp is null");
        }
        String localDataCenter = commonConfig.getDataCenter();
        if (genericResponse.isSuccess()) {
            updateState(genericResponse.getData());
            HeartBeatResponse heartBeatResponse = genericResponse.getData();
            centralControlLeaderExchanger.learn(localDataCenter, new Leader(heartBeatResponse.getMetaLeader(), heartBeatResponse.getMetaLeaderEpoch()));
            handleRenewResult(genericResponse.getData());
            renewFailCounter.set(0);
        } else {
            T data = genericResponse.getData();
            if (data == null) {
                // could no get data, trigger the counter inc
                throw new RuntimeException("renew node to metaServer error, resp.data is null, msg:" + genericResponse.getMessage());
            }
            // heartbeat on follow, refresh leader;
            // it will renewNode on leader next time;
            if (!data.isHeartbeatOnLeader()) {
                centralControlLeaderExchanger.learn(localDataCenter, new Leader(data.getMetaLeader(), data.getMetaLeaderEpoch()));
                // refresh the leader from follower, but the info maybe is incorrect
                // throw the exception to trigger the counter inc
                // if the info is correct, the counter would be reset after heartbeat
                throw new RuntimeException("renew node to metaServer.follower, leader is " + new Leader(data.getMetaLeader(), data.getMetaLeaderEpoch()));
            } else {
                throw new RuntimeException(StringFormatter.format("renew node to metaServer.leader error, msg={}, data={}", genericResponse.getMessage(), data));
            }
        }
    }

    void handleHeartbeatFailed(String leaderIp, Throwable e) {
        renewFailCounter.incrementAndGet();
        logger.error("[RenewNodeTask] renew node to metaServer error, fail count:{}, leader: {}", renewFailCounter.get(), leaderIp, e);
        throw new RuntimeException("renew node error!", e);
    }

    private void updateState(T response) {
        Map<String, Set<String>> map = response.getRemoteDataServers();
        if (!CollectionUtils.isEmpty(state.remoteDataServers)) {
            for (Entry<String, Set<String>> entry : state.remoteDataServers.entrySet()) {
                map.putIfAbsent(entry.getKey(), entry.getValue());
            }
        }
        State state1 = new State(response.getDataCentersFromMetaNodes(), response.getSessionNodesMap(), response.getSlots().getDataServers(), response.getSessionServerEpoch(), response.getMetaLeader(), response.getMetaLeaderEpoch(), map);
        this.state = state1;
        logger.info("update MetaStat, sessions={}/{}, datas={}, metaLeader: {}, metaLeaderEpoch: {}", state1.sessionServerEpoch, state1.clientInterfaceNodeToClientInterfaceNodeMap.keySet(), state1.dataServers, state1.metaLeader, state1.metaLeaderEpoch);
    }

    @Override
    public ProvideData getProvideData(String dataInfoId) {
        final String leaderIp = getMetaServerLeader();
        try {
            Response response = centralControlLeaderExchanger.sendRequest(commonConfig.getDataCenter(), new FetchProvideDataRequest(dataInfoId));

            Object result = response.getResult();
            if (result instanceof ProvideData) {
                return (ProvideData) result;
            } else {
                logger.error("fetch null provider data from {}", leaderIp);
                throw new RuntimeException("metaServerService fetch null provider data!");
            }
        } catch (Throwable e) {
            logger.error("fetch provider data error from {}", leaderIp, e);
            throw new RuntimeException("fetch provider data error! " + e.getMessage(), e);
        }
    }

    @Override
    public Map<String, ProvideData> getProvideData(Map<String, Long> dataInfoIdsWithVersion) {
        // TODO unsupported this now
        throw new UnsupportedOperationException();
    }

    @Override
    public FetchSystemPropertyResult fetchSystemProperty(String dataInfoId, long version) {
        final String leaderIp = getMetaServerLeader();

        try {
            Response response = centralControlLeaderExchanger.sendRequest(commonConfig.getDataCenter(), new FetchSystemPropertyRequest(dataInfoId, version));

            FetchSystemPropertyResult result = (FetchSystemPropertyResult) response.getResult();
            return result;
        } catch (Throwable e) {
            logger.error("fetch system property data:{}, version:{}, from {} is null", dataInfoId, version, leaderIp, e);
            throw new RuntimeException("fetch system property data error! " + e.getMessage(), e);
        }
    }

    @Override
    public SlotsStatusResponse getSlotsStatus() {
        final String leaderIp = getMetaServerLeader();

        try {
            Response response = centralControlLeaderExchanger.sendRequest(commonConfig.getDataCenter(), new GetSlotsStatusRequest());
            SlotsStatusResponse result = (SlotsStatusResponse) response.getResult();
            return result;
        } catch (Throwable e) {
            logger.error("fetch slot table status from={} error.", leaderIp, e);
            return null;
        }
    }

    @Override
    public Set<ClientInterfaceId> getClientInterfaceIdSet() {
        Set<ClientInterfaceId> set = new HashSet<>(state.clientInterfaceNodeToClientInterfaceNodeMap.size());
        for (ClientInterfaceNode clientInterfaceNode : state.clientInterfaceNodeToClientInterfaceNodeMap.values()) {
            set.add(clientInterfaceNode.getClientInterfaceId());
        }
        return set;
    }

    public Map<String, ClientInterfaceNode> getSessionNodes() {
        return state.clientInterfaceNodeToClientInterfaceNodeMap;
    }

    public Set<String> getSessionServerList() {
        return state.clientInterfaceNodeToClientInterfaceNodeMap.keySet();
    }

    public Set<String> getDataServerList() {
        return state.dataServers;
    }

    public Map<String, Set<String>> getRemoteDataServers() {
        return state.remoteDataServers;
    }

    public String getMetaServerLeader() {
        String localDataCenter = commonConfig.getDataCenter();
        Leader leader = centralControlLeaderExchanger.getLeader(localDataCenter);
        if (leader == null) {
            throw new RuntimeException("localDataCenter meta leader is null.");
        }
        return leader.getLeader();
    }

    public List<String> getSessionServerList(String zonename) {
        List<String> serverList = new ArrayList<>();
        for (ClientInterfaceNode clientInterfaceNode : getSessionNodes().values()) {
            if (StringUtils.isBlank(zonename) || zonename.equals(clientInterfaceNode.getRegionId())) {
                Url url = clientInterfaceNode.getUrl();
                if (url != null) {
                    serverList.add(url.getIp());
                }
            }
        }
        return serverList;
    }

    public List<ClientInterfaceNode> getSessionNodeWithConnNumList(String zonename) {
        List<ClientInterfaceNode> serverList = new ArrayList<>();
        for (ClientInterfaceNode clientInterfaceNode : getSessionNodes().values()) {
            if (StringUtils.isBlank(zonename) || zonename.equals(clientInterfaceNode.getRegionId())) {
                Url url = clientInterfaceNode.getUrl();
                if (url != null) {
                    serverList.add(clientInterfaceNode);
                }
            }
        }
        return serverList;
    }

    @Override
    public long getSessionServerEpoch() {
        return state.sessionServerEpoch;
    }

    @Override
    public Set<String> getDataCenters() {
        return state.dataCenters;
    }

    protected abstract void handleRenewResult(T result);

    protected abstract HeartbeatRequest createRequest();

    protected abstract NodeType nodeType();

    protected abstract String cell();

    protected abstract long getCurrentslotsEpoch();


    private static final class State {
        static final State NULL = new State(Collections.emptySet(), Collections.emptyMap(), Collections.emptySet(), 0, null, -1L, Collections.emptyMap());
        protected final long sessionServerEpoch;
        protected final Map<String, ClientInterfaceNode> clientInterfaceNodeToClientInterfaceNodeMap;
        protected final Set<String> dataServers;
        protected final String metaLeader;
        protected final long metaLeaderEpoch;
        protected final Set<String> dataCenters;
        protected final Map<String, Set<String>> remoteDataServers;

        State(Set<String> dataCenters, Map<String, ClientInterfaceNode> clientInterfaceNodeToClientInterfaceNodeMap, Set<String> dataServers, long sessionServerEpoch, String metaLeader, long metaLeaderEpoch, Map<String, Set<String>> remoteDataServers) {
            this.sessionServerEpoch = sessionServerEpoch;
            this.dataCenters = Collections.unmodifiableSet(new TreeSet<>(dataCenters));
            this.clientInterfaceNodeToClientInterfaceNodeMap = Collections.unmodifiableMap(clientInterfaceNodeToClientInterfaceNodeMap);
            this.dataServers = Collections.unmodifiableSet(dataServers);
            this.metaLeader = metaLeader;
            this.metaLeaderEpoch = metaLeaderEpoch;
            this.remoteDataServers = remoteDataServers;
        }
    }

    private final class Renewer extends WakeUpLoopExecuteTask {
        @Override
        public void _execute_() {
            try {
                checkRenewFailCounter();
                renewNode();
            } catch (Throwable e) {
                logger.error("failed to renewNode", e);
            }
        }

        @Override
        public int getWaitingMillis() {
            return getRenewIntervalSecs() * 1000;
        }
    }
}
